package com.haozhen.homework

/**
  * @author haozhen
  * @email haozh@ync1.com
  * @date 2021/1/3  16:28
  */
object Work4 {

  def calculateDisttance(x1 :Double,x2:Double,x3:Double,x4:Double,
                         y1 :Double,y2:Double,y3:Double,y4:Double): Double ={
       Math.sqrt(math.pow((x1-y1),2)+math.pow((x2-y2),2)+
        math.pow((x3-y3),2)+math.pow((x4-y4),2))
  }

  def main(args: Array[String]): Unit = {
    import org.apache.spark.sql.SparkSession

    val spark = SparkSession.builder()
      .appName("work4")
      .master("local[*]")
      .getOrCreate()
    spark.sparkContext.setLogLevel("WARN")

    val k = 5;

    var modleData = spark.read
      .option("header","true")
      .option("delimiter",",")
      .option("inferschema","true")
      .csv("data/Iris.csv")

    var testData = spark.read
      .option("header","true")
      .option("delimiter",",")
      .option("inferschema","true")
      .csv("data/Iris_test.csv")

    val carData = modleData.rdd.cartesian(testData.rdd)

    val mapData = carData.map {
      case (x, y) => {
        val xid = x.getInt(0)
        val yid = y.getInt(0)
        val irisType = x.getString(5)
        val distance = calculateDisttance(
          x.getDouble(1), x.getDouble(2), x.getDouble(3), x.getDouble(4),
          y.getDouble(1), y.getDouble(2), y.getDouble(3), y.getDouble(4)
        )
        (yid, (distance, irisType))
      }
    }


    mapData.groupByKey().mapValues(v=>{
      val tuples = v.toList.sortBy(_._1).take(k)

      val stringToTuples = tuples.groupBy(_._2)

      val strings = stringToTuples.toList.sortBy(_._2.size).last._2.map(_._2)
      strings(0)
    }).foreach(x=>println(s"id: ${x._1},种类：${x._2}"))

    spark.close()
  }

}
